﻿using NPOI.OpenXmlFormats.Dml.Chart;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace Publicuse.Util.RabbitMQ
{
    /// <summary>
    /// RabbitMQ工具类
    /// </summary>
    public class RabbitMQUtil
    {

        /// <summary>
        /// 初始化连接
        /// </summary>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        private static void Init(out string msg, Action<IModel> action)
        {
            msg = string.Empty;
            try
            {
                //创建连接工厂
                var factory = new ConnectionFactory();
                factory.HostName = AppSettingUtil.RabbitMQHost; //服务地址
                factory.UserName = AppSettingUtil.RabbitMQName;//用户名
                factory.Password = AppSettingUtil.RabbitMQPass;//密码
                //创建连接
                using (var connection = factory.CreateConnection())
                {
                    //创建通道
                    using (var channel = connection.CreateModel())
                    {
                        action.Invoke(channel);
                    }
                }
            }
            catch (Exception ex)
            {
                msg = ex.Message;
            }
        }

        /// <summary>
        /// 生产者消费者 一个任务一个消费
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="exchange"></param>
        /// <param name="data"></param>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        public static void CreateTask(string queue, string exchange, string data, out string msg,Action<string,string> action)
        {
            msg = string.Empty;
            Init(out msg, channel =>
            {
                //声明一个队列
                channel.QueueDeclare(queue: queue, durable: true, exclusive: false, autoDelete: false, arguments: null);
                //声明交换机exchang
                channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Direct, durable: true, autoDelete: false, arguments: null);
                //绑定声明交换机和队列
                channel.QueueBind(queue: queue, exchange: exchange, routingKey: string.Empty, arguments: null);
                byte[] body = Encoding.UTF8.GetBytes(data);
                //开启消息确认模式
                channel.ConfirmSelect();
                //发布消息
                channel.BasicPublish(exchange: exchange,
                                      routingKey: string.Empty,
                                      basicProperties: null,
                                      body: body);

                //事务提交
                if (channel.WaitForConfirms()) //如果一条消息或多消息都确认发送
                {
                    Console.ForegroundColor = ConsoleColor.Red;
                    action?.Invoke(queue, data);
                }
                else
                {
                    //肯定需要来重新写入消息；
                    CreateTask(queue,exchange,data,out var msg, action);
                }
                channel.WaitForConfirmsOrDie();//如果所有消息发送成功 就正常执行；如果有消息发送失败；就抛出异常；
            });
        }

        /// <summary>
        /// 订阅模式一对多一个消息可以多个消费
        /// </summary>
        /// <param name="queues"></param>
        /// <param name="exchange"></param>
        /// <param name="data"></param>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        public static void CreateTask(List<string> queues, string exchange, string data, out string msg, Action<string,string> action)
        {
            msg = string.Empty;
            Init(out msg, channel =>
            {
                //声明交换机exchang
                channel.ExchangeDeclare(exchange,ExchangeType.Fanout,true,false,null);
                foreach (var item in queues)
                {
                    //声明一个队列
                    channel.QueueDeclare(queue: item, durable: true, exclusive: false, autoDelete: false, arguments: null);
                    //绑定声明交换机和队列
                    channel.QueueBind(queue: item, exchange: exchange, routingKey: string.Empty, arguments: null);

                    byte[] body = Encoding.UTF8.GetBytes(data);
                    Console.ForegroundColor = ConsoleColor.Red;
                    //开启消息确认模式
                    channel.ConfirmSelect();
                    channel.BasicPublish(exchange: exchange,
                                   routingKey: string.Empty,
                                   basicProperties: null,
                                   body: body);
                    //事务提交
                    if (channel.WaitForConfirms()) //如果一条消息或多消息都确认发送
                    {
                        Console.ForegroundColor = ConsoleColor.Red;
                        action?.Invoke(item, data);
                    }
                    else
                    {
                        //肯定需要来重新写入消息；
                        CreateTask(queues, exchange, data, out var msg, action);
                    }
                    channel.WaitForConfirmsOrDie();//如果所有消息发送成功 就正常执行；如果有消息发送失败；就抛出异常；
                }
            });
        }

        /// <summary>
        /// 生产者消费者
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        public static void ReadTask(string queue, out string msg, Func<string,string,bool> action)
        {
            msg = string.Empty;
            Init(out msg, channel =>
            {
                Console.ForegroundColor = ConsoleColor.Green;
                //事件基本消费者
                var consumer = new EventingBasicConsumer(channel);
                //接收到消息事件
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var data = Encoding.UTF8.GetString(body.ToArray());
                    Console.ForegroundColor = ConsoleColor.Green;
                    try
                    {
                        var result = (bool)action?.Invoke(queue, data);
                        if (result)
                        {
                            //手动确认  消息正常消费  告诉Broker：你可以把当前这条消息删除掉了
                            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                        }
                        else
                        {
                            //否定：告诉Broker，这个消息我灭有正常消费；  requeue: true：重新写入到队列里去； false:你还是删除掉；
                            channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
                        }
                    }
                    catch (Exception ex)
                    {
                        //否定：告诉Broker，这个消息我灭有正常消费；  requeue: true：重新写入到队列里去； false:你还是删除掉；
                        channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
                    }
                };
                //启动消费者 设置为手动应答消息
                channel.BasicConsume(queue: queue,
                             autoAck: false,
                             consumer: consumer);
                Console.Read();
            });
        }

        /// <summary>
        /// 发布订阅
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="exchange"></param>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        public static void ReadTask(string queue, string exchange,out string msg, Func<string,string,bool> action)
        {
            msg = string.Empty;
            Init(out msg, channel =>
            {
                channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
                channel.QueueDeclare(queue: queue, durable: true, exclusive: false, autoDelete: false, arguments: null);
                channel.QueueBind(queue: queue, exchange: exchange, routingKey: string.Empty, arguments: null);
                try
                {
                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var body = ea.Body;
                        var data = Encoding.UTF8.GetString(body.ToArray());
                        try
                        {
                            var result = (bool)action?.Invoke(queue, data);
                            if (result)
                            {
                                //手动确认  消息正常消费  告诉Broker：你可以把当前这条消息删除掉了
                                channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                            }
                            else
                            {
                                //否定：告诉Broker，这个消息我灭有正常消费；  requeue: true：重新写入到队列里去； false:你还是删除掉；
                                channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
                            }
                        }
                        catch (Exception ex)
                        {
                            //否定：告诉Broker，这个消息我灭有正常消费；  requeue: true：重新写入到队列里去； false:你还是删除掉；
                            channel.BasicReject(deliveryTag: ea.DeliveryTag, requeue: true);
                        }
                       
                    };
                    channel.BasicConsume(queue: queue, autoAck: false, consumer: consumer);
                    Console.Read();
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                Console.Read();
            });
        }

        /// <summary>
        /// 发布订阅
        /// </summary>
        /// <param name="queue"></param>
        /// <param name="exchange"></param>
        /// <param name="data"></param>
        /// <param name="msg"></param>
        public static void ReadTask(string queue, string exchange, string data, out string msg)
        {
            msg = string.Empty;
            Init(out msg, channel =>
            {
                channel.ExchangeDeclare(exchange: exchange, type: ExchangeType.Fanout, durable: true, autoDelete: false, arguments: null);
                channel.QueueDeclare(queue: queue, durable: true, exclusive: false, autoDelete: false, arguments: null);
                channel.QueueBind(queue: queue, exchange: exchange, routingKey: string.Empty, arguments: null);
                byte[] body = Encoding.UTF8.GetBytes(data);
                channel.BasicPublish(exchange: exchange,
                                       routingKey: string.Empty,
                                       basicProperties: null,
                                       body: body);
                Console.Read();
            });
        }

    }
}
